import torch
import torch.utils.data
from torch import nn, optim
from torch.nn import functional as F
from torchvision import datasets,transforms
from torchvision.utils import save_image, make_grid
from PIL import Image
# from tqdm.notebook import tqdm
from tqdm import tqdm
# from tqdm.auto import tqdm
import os
import shutil
import random
random.seed(5)
import matplotlib
%matplotlib inline
%config InlineBackend.figure_format='svg'
import matplotlib.pyplot as plt
plt.style.use('ggplot')
import numpy as np
# Dataset Attr2font
from datetime import datetime

from torch.utils.tensorboard import SummaryWriter

## Conexão com o google drive, se achar necessário
#from google.colab import drive
#drive.mount('/content/drive', force_remount=True)

# Baixando os dados do Attr2font que podem ser encontrados em:
# !gdown --id 1TTqAklfsAp6KOPxCVl2jktH8kN4lEmI_
# !unzip ./_data/explor_all.zip

# Cria os diretórios de interesse:
# !mkdir _data
# !mkdir results
# !mkdir models
# !mv ./explor_all/image ./_data/Fonts
# !rm -rf ./explor_all

def count_folders_and_png_files(folder, msg):
    items = os.listdir(folder)
    folders = [item for item in items if os.path.isdir(os.path.join(folder, item))]
    print(f"{msg} number of subfolders: {len(folders):,}")

    png_count = 0
    for _, _, files in os.walk(folder):
        png_files = [file for file in files if file.lower().endswith('.png')]
        png_count += len(png_files)
    print(f"{msg} number of png files: {png_count:,}")

count_folders_and_png_files('./_data/Fonts', "'Fonts'")

'Fonts' number of subfolders: 1,116
'Fonts' number of png files: 69,192

epochs=5
batch_size=64
torch.manual_seed(17)

<torch._C.Generator at 0x7f9d281cd670>

def show_img(img, title=None):
    img = img.permute(1, 2, 0)
    if img.shape[2]==1:
        img = img.view(img.shape[0], img.shape[1])
    if title is None:
        plt.title(f'Image has size {img.cpu().numpy().shape}')
    else:
        plt.title(title)
    plt.imshow(img,cmap='gray')
    plt.axis('off')
    plt.show()

path='./_data/Fonts'
train_path = './_data/trainFonts'
val_path = './_data/valFonts'

for f in [path, train_path, val_path]:
    os.makedirs(f, exist_ok=True)

transforms_all = transforms.Compose([
    transforms.Grayscale(num_output_channels=1),
    transforms.Resize(size=50),
    transforms.ToTensor()
])

dummy_batch = torch.utils.data.DataLoader(
    datasets.ImageFolder(path,transform=transforms.ToTensor()),
    batch_size=1, shuffle=True
)

for batch in dummy_batch:
  original_image=batch[0][0]
  show_img(original_image)
  show_img(transforms_all(transforms.ToPILImage()(original_image)))
  break

folders=os.listdir(path)
count_folders_and_png_files(path, "Original Fonts")

shutil.rmtree(train_path)
shutil.rmtree(val_path)

os.makedirs(train_path, exist_ok=True)
os.makedirs(val_path, exist_ok=True)

# Podemos limitar em num_folders a quantidade total de pasta que iremos trabalhar, no total temos 1116.
num_folders = 134
count_folders = 1
for folder in folders:
    if count_folders <= num_folders:
      os.mkdir(f'{train_path}/{folder}')
      os.mkdir(f'{val_path}/{folder}')
      images = os.listdir(f'{path}/{folder}')
      random.shuffle(images)
      for image in images[:20]:
        shutil.copy(f'{path}/{folder}/{image}', f'{val_path}/{folder}/{image}')
      for image in images[20:]:
        shutil.copy(f'{path}/{folder}/{image}', f'{train_path}/{folder}/{image}')
    count_folders += 1

count_folders_and_png_files(train_path, "Train Fonts")
count_folders_and_png_files(val_path, "Val Fonts")

Original Fonts number of subfolders: 1,116
Original Fonts number of png files: 69,192
Train Fonts number of subfolders: 134
Train Fonts number of png files: 5,628
Val Fonts number of subfolders: 134
Val Fonts number of png files: 2,680

print(len(os.listdir(train_path)))
print(len(os.listdir(val_path)))

134
134

def get_dataloaders(batch_size=64, size=50):
    transforms_set = transforms.Compose([
        transforms.Grayscale(num_output_channels=1),
        transforms.Resize(size=size),
        transforms.ToTensor()
    ])

    train_loader=torch.utils.data.DataLoader(
        datasets.ImageFolder(f'{train_path}',transform=transforms_set),
        batch_size=batch_size, shuffle=True
    )

    val_loader=torch.utils.data.DataLoader(
        datasets.ImageFolder(f'{val_path}',transform=transforms_set),
        batch_size=batch_size, shuffle=True
    )

    return train_loader, val_loader

class VAE(nn.Module):
    def __init__(self, input_dim=50*50, z=32, hidden_dim=1000, n_hidden_encoder=1, n_hidden_decoder=1):
        super(VAE, self).__init__()

        assert n_hidden_encoder >= 1, 'Number of hidden layers must be at least 1.'
        assert n_hidden_decoder >= 1, 'Number of hidden layers must be at least 1.'

        self.input_dim = input_dim # Dimensão de entrada
        self.z = z # Dimensão do espaço latente
        self.hidden_dim = hidden_dim # Dimensão da camada escondida
        self.n_hidden_encoder = n_hidden_encoder # Número de camadas escondidas no encoder
        self.n_hidden_decoder = n_hidden_decoder # Número de camadas escondidas no decoder

        #ENCODER
        self.encoder = nn.Sequential()
         # Camada densa do dado de entrada para dimensão hidden_dim (1000)
        self.encoder.add_module('input_linear', nn.Linear(input_dim,hidden_dim))
        self.encoder.add_module('input_act',nn.ReLU())
        # Camadas densas ocultas adicionais
        for i in range(1, n_hidden_encoder):
            self.encoder.add_module(f'{i+1}_linear',nn.Linear(hidden_dim, hidden_dim))
            self.encoder.add_module(f'{i+1}_act',nn.ReLU())
        # Camadas densas de dimensão hidden_dim para espaço latente z: médias e log(vars)
        self.fc21 = nn.Linear(hidden_dim,z)
        self.fc22 = nn.Linear(hidden_dim,z)

        #DECODER
        self.decoder = nn.Sequential()
        # Camada densa do espaço latente z para dimensão hidden_dim
        self.decoder.add_module('input_linear', nn.Linear(z,hidden_dim))
        self.decoder.add_module('input_act',nn.ReLU())
        # Camadas densas ocultas adicionais
        for i in range(1, n_hidden_decoder):
            self.decoder.add_module(f'{i+1}_linear',nn.Linear(hidden_dim, hidden_dim))
            self.decoder.add_module(f'{i+1}_act',nn.ReLU())
        # Camadas densas de dimensão hidden_dim para dimensão de entrada
        self.decoder.add_module('output_linear', nn.Linear(hidden_dim,input_dim))
        self.decoder.add_module('output_act',nn.Sigmoid())

    def encode(self, x):
        h1 = self.encoder(x)
        return self.fc21(h1), self.fc22(h1)

    def reparameterize(self, mu, logvar):
        std = torch.exp(0.5*logvar)
        eps = torch.randn_like(std)
        return mu + eps*std

    def decode(self, z):
        return self.decoder(z)

    def forward(self, x):
        # don't forget forward pass re-index
        mu, logvar = self.encode(x.view(-1, self.input_dim))
        z = self.reparameterize(mu, logvar)
        return self.decode(z), mu, logvar

def loss_function(recon_x, x, mu, logvar, beta=0.50, input_dim=50*50):
    BCE = F.binary_cross_entropy(recon_x, x.view(-1, input_dim), reduction='sum')
    KLD = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp())
    loss = 2.0 * ((1 - beta) * BCE + beta * KLD)
    return loss, BCE.detach(), KLD.detach()

def initialize_test_run(options):
    device=torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    print('Device: ',device)
    model=VAE(
        input_dim=options['input_dim_flat'],
        z=options['z'],
        hidden_dim=options['hidden_dim'],
        n_hidden_encoder=options['n_hidden_encoder'],
        n_hidden_decoder=options['n_hidden_decoder']).to(device)
    optimizer=optim.Adam(model.parameters(),lr=options['lr'])
    writer = SummaryWriter(f'_runs/{options["model_name"]}')
    return device, model, optimizer, writer

def evaluate(epoch, device, model, writer, val_loader, options):

    val_loss = 0
    val_loss_bce = 0
    val_loss_kld = 0
    with torch.no_grad():
        for i, (data, _) in enumerate(val_loader):

            data=data.to(device)
            recon_batch,mu,logvar=model(data)
            loss, loss_bce, loss_kld = loss_function(recon_batch,data,mu,logvar)
            val_loss += loss.item()
            val_loss_bce += loss_bce.item()
            val_loss_kld += loss_kld.item()

            if i == 0 and epoch%50==1:
                n = min(data.size(0), 16)
                comparison = torch.cat([
                    data[:n],
                    recon_batch.view(options['batch_size'], 1,
                                     options['input_dim'][0], options['input_dim'][1])[:n]]).cpu()
                grid = make_grid(comparison, nrow=16, normalize=True, scale_each=True).cpu()
                show_img(grid, title=f'Epoch {epoch}')
                writer.add_images('Img/Reconstruction', grid, epoch, dataformats='CHW')

    val_loss /= len(val_loader.dataset)
    val_loss_bce /= len(val_loader.dataset)
    val_loss_kld /= len(val_loader.dataset)
    return val_loss, val_loss_bce, val_loss_kld


def sample_latent_space(epoch, device, model, writer, options):
    with torch.no_grad():

        # how to sample from our latent space
        sample=torch.randn(64,32).to(device)
        sample=model.decode(sample).cpu()

        grid = make_grid(sample.view(64, 1, options['input_dim'][0], options['input_dim'][1]), nrow=8, normalize=False, scale_each=False)
        show_img(grid, title=f'Epoch {epoch}')
        writer.add_images('Img/Samples', grid, epoch, dataformats='CHW')

def train(device, model, optimizer, writer, train_loader, val_loader, epoch, options):
    model.train()

    train_loss = 0
    train_loss_bce = 0
    train_loss_kld = 0

    progress_bar = tqdm(train_loader, desc=f'Epoch {epoch:03d}', leave=False, disable=False)
    for data, _ in progress_bar:
        progress_bar.set_description(f'Epoch {epoch:03d}')

        data=data.to(device)
        optimizer.zero_grad()
        recon_batch,mu,logvar=model(data)

        loss, loss_bce, loss_kld = loss_function(
            recon_batch,data,mu,logvar, beta=options['beta'], input_dim=options['input_dim_flat'])
        loss.backward()
        train_loss += loss.item()
        train_loss_bce += loss_bce.item()
        train_loss_kld += loss_kld.item()
        optimizer.step()

        progress_bar.set_postfix({'training_loss': f'{(loss.item() / len(data)):.3f}'})
    progress_bar.close()

    avg_train_loss = train_loss / len(train_loader.dataset)
    avg_train_loss_bce = train_loss_bce / len(train_loader.dataset)
    avg_train_loss_kld = train_loss_kld / len(train_loader.dataset)
    val_loss, val_loss_bce, val_loss_kld = evaluate(
        epoch, device, model, writer, val_loader, options)

    print(f'Epoch {epoch:03d}: train={avg_train_loss:<8.3f} (BCE={avg_train_loss_bce:<8.3f}, KLD={avg_train_loss_kld:<8.3f}), val: {val_loss:<8.3f}')

    writer.add_scalar('Train_Loss/Total', avg_train_loss, epoch)
    writer.add_scalar('Train_Loss/BCE', avg_train_loss_bce, epoch)
    writer.add_scalar('Train_Loss/KLD', avg_train_loss_kld, epoch)

    writer.add_scalar('Val_Loss/Total', val_loss, epoch)
    writer.add_scalar('Val_Loss/BCE', val_loss_bce, epoch)
    writer.add_scalar('Val_Loss/KLD', val_loss_kld, epoch)

    if epoch%100==0:
        torch.save(model.state_dict(), f'./_models/{options["model_name"]}_{epoch}.model')

def make_run(options):
    print(options['model_name'])

    options['input_dim_flat'] = options['input_dim'][0] * options['input_dim'][1]
    device, model, optimizer, writer = initialize_test_run(options)
    train_loader, val_loader = get_dataloaders(options['batch_size'])

    writer.add_hparams({k:str(v) for k,v in options.items()}, {})

    epochs = options['epochs']
    for epoch in range(1,epochs+1):
        train(device, model, optimizer, writer, train_loader, val_loader, epoch, options)
        if epoch%50==1:
            sample_latent_space(epoch, device, model, writer, options)

    writer.flush()
    writer.close()

model_name = f'vae_{datetime.now().strftime("%Y%m%d-%H%M%S")}'
options1 = {
    'model_name': model_name,
    'input_dim': (50, 50),
    'z': 32,
    'hidden_dim': 1000,
    'n_hidden_encoder': 1,
    'n_hidden_decoder': 1,
    'lr': 1e-3,
    'beta': 0.50,
    'epochs': 500,
    'batch_size': 64
}

make_run(options1)

vae_20240927-171921
Device:  cuda

Epoch 001: train=676.816  (BCE=649.598 , KLD=27.218  ), val: 537.187

Epoch 002: train=492.535  (BCE=463.625 , KLD=28.910  ), val: 452.310

Epoch 003: train=431.954  (BCE=397.517 , KLD=34.437  ), val: 411.958

Epoch 004: train=397.038  (BCE=360.018 , KLD=37.021  ), val: 383.824

model_name = f'vae_{datetime.now().strftime("%Y%m%d-%H%M%S")}'
options2 = {k:v for k,v in options1.items()}
options2['model_name'] = model_name
options2['beta'] = 0.80
make_run(options2)

vae_20240927-180814
Device:  cuda

Epoch 001:   0%|          | 0/88 [00:00<?, ?it/s]

Epoch 001: train=295.979  (BCE=711.400 , KLD=7.137   ), val: 593.248

Epoch 002: train=232.339  (BCE=545.885 , KLD=8.741   ), val: 531.879

Epoch 003: train=217.613  (BCE=504.182 , KLD=9.962   ), val: 489.166

Epoch 004: train=203.030  (BCE=458.367 , KLD=12.302  ), val: 453.492

model_name = f'vae_{datetime.now().strftime("%Y%m%d-%H%M%S")}'
options3 = {k:v for k,v in options2.items()}
options3['model_name'] = model_name
options3['n_hidden_encoder'] = 2
make_run(options3)

vae_20240927-185653
Device:  cuda

Epoch 001: train=295.441  (BCE=715.289 , KLD=5.828   ), val: 619.177

Epoch 002: train=240.465  (BCE=570.784 , KLD=7.595   ), val: 546.042

Epoch 003: train=223.740  (BCE=521.464 , KLD=9.471   ), val: 515.833

Epoch 004: train=216.838  (BCE=500.434 , KLD=10.415  ), val: 498.716

model_list = [
    ('vae_20240927-171921_500.model', options1),
    ('vae_20240927-180814_500.model', options2),
    ('vae_20240927-185653_500.model', options3)
]

def load_model(model_name, options):
    options['input_dim_flat'] = options['input_dim'][0] * options['input_dim'][1]
    device, model, _, _ = initialize_test_run(options)
    model.load_state_dict(
        torch.load(f'./_models/{model_name}',
                   map_location=torch.device('cpu'),
                   weights_only=False))
    return model, device

models = [load_model(model_name, options) for model_name, options in model_list]

Device:  cuda
Device:  cuda
Device:  cuda

def get_samples(model, device, latent_dim=32, n_samples=64):
    with torch.no_grad():
        sample=torch.randn(n_samples,latent_dim).to(device)
        return model.decode(sample).cpu().view(n_samples, 1, 50, 50)

def show_samples(samples, nrow=3, title=None):
    grid = make_grid(samples, nrow=nrow, normalize=False, scale_each=False)
    show_img(grid, title=title)

for i, (m, d) in enumerate(models):
    samples = get_samples(m, d, latent_dim=32, n_samples=16)
    show_samples(samples, nrow=4, title=f"Model #{i+1}")

#Implementação de https://www.kaggle.com/code/ibtesama/gan-in-pytorch-with-fid
from scipy import linalg
from torch.nn.functional import adaptive_avg_pool2d
import torchvision.models as tvision_models

class InceptionV3(nn.Module):
    """Pretrained InceptionV3 network returning feature maps"""

    # Index of default block of inception to return,
    # corresponds to output of final average pooling
    DEFAULT_BLOCK_INDEX = 3

    # Maps feature dimensionality to their output blocks indices
    BLOCK_INDEX_BY_DIM = {
        64: 0,   # First max pooling features
        192: 1,  # Second max pooling featurs
        768: 2,  # Pre-aux classifier features
        2048: 3  # Final average pooling features
    }

    def __init__(self,
                 output_blocks=[DEFAULT_BLOCK_INDEX],
                 resize_input=True,
                 normalize_input=True,
                 requires_grad=False):

        super(InceptionV3, self).__init__()

        self.resize_input = resize_input
        self.normalize_input = normalize_input
        self.output_blocks = sorted(output_blocks)
        self.last_needed_block = max(output_blocks)

        assert self.last_needed_block <= 3, \
            'Last possible output block index is 3'

        self.blocks = nn.ModuleList()


        inception = tvision_models.inception_v3(weights=True)

        # Block 0: input to maxpool1
        block0 = [
            inception.Conv2d_1a_3x3,
            inception.Conv2d_2a_3x3,
            inception.Conv2d_2b_3x3,
            nn.MaxPool2d(kernel_size=3, stride=2)
        ]
        self.blocks.append(nn.Sequential(*block0))

        # Block 1: maxpool1 to maxpool2
        if self.last_needed_block >= 1:
            block1 = [
                inception.Conv2d_3b_1x1,
                inception.Conv2d_4a_3x3,
                nn.MaxPool2d(kernel_size=3, stride=2)
            ]
            self.blocks.append(nn.Sequential(*block1))

        # Block 2: maxpool2 to aux classifier
        if self.last_needed_block >= 2:
            block2 = [
                inception.Mixed_5b,
                inception.Mixed_5c,
                inception.Mixed_5d,
                inception.Mixed_6a,
                inception.Mixed_6b,
                inception.Mixed_6c,
                inception.Mixed_6d,
                inception.Mixed_6e,
            ]
            self.blocks.append(nn.Sequential(*block2))

        # Block 3: aux classifier to final avgpool
        if self.last_needed_block >= 3:
            block3 = [
                inception.Mixed_7a,
                inception.Mixed_7b,
                inception.Mixed_7c,
                nn.AdaptiveAvgPool2d(output_size=(1, 1))
            ]
            self.blocks.append(nn.Sequential(*block3))

        for param in self.parameters():
            param.requires_grad = requires_grad

    def forward(self, inp):
        """Get Inception feature maps
        Parameters
        ----------
        inp : torch.autograd.Variable
            Input tensor of shape Bx3xHxW. Values are expected to be in
            range (0, 1)
        Returns
        -------
        List of torch.autograd.Variable, corresponding to the selected output
        block, sorted ascending by index
        """
        outp = []
        x = inp

        if self.resize_input:
            x = F.interpolate(x,
                              size=(299, 299),
                              mode='bilinear',
                              align_corners=False)

        if self.normalize_input:
            x = 2 * x - 1  # Scale from range (0, 1) to range (-1, 1)

        for idx, block in enumerate(self.blocks):
            x = block(x)
            if idx in self.output_blocks:
                outp.append(x)

            if idx == self.last_needed_block:
                break

        return outp

block_idx = InceptionV3.BLOCK_INDEX_BY_DIM[2048]
model_inceptionv3 = InceptionV3(output_blocks=[block_idx])
model_inceptionv3=model_inceptionv3.cuda()

def calculate_activation_statistics(images,model,batch_size=128, dims=2048,
                    cuda=False):
    model.eval()
    act=np.empty((len(images), dims))

    if cuda:
        batch=images.cuda()
    else:
        batch=images
    pred = model(batch)[0]

    # If model output is not scalar, apply global spatial average pooling.
    # This happens if you choose a dimensionality not equal 2048.
    if pred.size(2) != 1 or pred.size(3) != 1:
        pred = adaptive_avg_pool2d(pred, output_size=(1, 1))

    act = pred.cpu().data.numpy().reshape(pred.size(0), -1)

    mu = np.mean(act, axis=0)
    sigma = np.cov(act, rowvar=False)
    return mu, sigma

def calculate_frechet_distance(mu1, sigma1, mu2, sigma2, eps=1e-6):
    """Numpy implementation of the Frechet Distance.
    The Frechet distance between two multivariate Gaussians X_1 ~ N(mu_1, C_1)
    and X_2 ~ N(mu_2, C_2) is
            d^2 = ||mu_1 - mu_2||^2 + Tr(C_1 + C_2 - 2*sqrt(C_1*C_2)).
    """

    mu1 = np.atleast_1d(mu1)
    mu2 = np.atleast_1d(mu2)

    sigma1 = np.atleast_2d(sigma1)
    sigma2 = np.atleast_2d(sigma2)

    assert mu1.shape == mu2.shape, \
        'Training and test mean vectors have different lengths'
    assert sigma1.shape == sigma2.shape, \
        'Training and test covariances have different dimensions'

    diff = mu1 - mu2

    covmean, _ = linalg.sqrtm(sigma1.dot(sigma2), disp=False)
    if not np.isfinite(covmean).all():
        msg = ('fid calculation produces singular product; '
               'adding %s to diagonal of cov estimates') % eps
        print(msg)
        offset = np.eye(sigma1.shape[0]) * eps
        covmean = linalg.sqrtm((sigma1 + offset).dot(sigma2 + offset))


    if np.iscomplexobj(covmean):
        if not np.allclose(np.diagonal(covmean).imag, 0, atol=1e-3):
            m = np.max(np.abs(covmean.imag))
            raise ValueError('Imaginary component {}'.format(m))
        covmean = covmean.real

    tr_covmean = np.trace(covmean)

    return (diff.dot(diff) + np.trace(sigma1) +
            np.trace(sigma2) - 2 * tr_covmean)

def calculate_fretchet(images_real,images_fake,model):
     mu_1,std_1=calculate_activation_statistics(images_real,model,cuda=True)
     mu_2,std_2=calculate_activation_statistics(images_fake,model,cuda=True)

     """get fretched distance"""
     fid_value = calculate_frechet_distance(mu_1, std_1, mu_2, std_2)
     return fid_value

n_images = 2**7

_, val_loader = get_dataloaders(n_images)

real_images = next(iter(val_loader))[0].repeat(1, 3, 1, 1)
show_samples(real_images[:16], nrow=4, title='Real images')

i = 0
m = models[i][0]
d = models[i][1]
samples = get_samples(m, d, latent_dim=32, n_samples=1000).repeat(1, 3, 1, 1)
fid = calculate_fretchet(real_images, samples, model_inceptionv3)
show_samples(samples[:16], nrow=4, title=f'Model {i+1}: FID={fid:0.4e}')

i = 1
m = models[i][0]
d = models[i][1]
samples = get_samples(m, d, latent_dim=32, n_samples=1000).repeat(1, 3, 1, 1)
fid = calculate_fretchet(real_images, samples, model_inceptionv3)
show_samples(samples[:16], nrow=4, title=f'Model {i+1}: FID={fid:0.4e}')

i = 2
m = models[i][0]
d = models[i][1]
samples = get_samples(m, d, latent_dim=32, n_samples=1000).repeat(1, 3, 1, 1)
fid = calculate_fretchet(real_images, samples, model_inceptionv3)
show_samples(samples[:16], nrow=4, title=f'Model {i+1}: FID={fid:0.4e}')

Attr2Font¶

Pacotes¶

Google Drive¶

Dados:¶

Separação em dados de treino e validação¶

Construção do Dataset e Dataloader¶

Implementação VAE¶

Implementação Loss Function¶

Criação do Modelo¶

Loops de Treinamento e Avaliação¶

Treinamento¶

Qualidade das Amostras Geradas¶